package org.example.MqttDemo;

import org.eclipse.paho.client.mqttv3.*;
import org.example.obj.MqttConfiguration;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public class MqttConnect  {
//  Configuration class
    MqttConfiguration MQ;
    MqttConnectOptions MO;
//    client project
    private  MqttClient client ;
//     The length is limited to 64k but within that there are no limits to the number of levels in a topic tree.


    /**
     *  构造类
     * @param MQ MQTT配置类
     */
    public MqttConnect(MqttConfiguration MQ ) {
            this.MQ = MQ;

        System.out.println(MQ);
    }
    public MqttConnect(){}
    public boolean status()   {
            if (MQ == null) return false;
        try {
            String url = MQ.getBrokerUrl()+":"+MQ.getPort();
            client  = new MqttClient(url,MQ.getClientID());
            client.connect(MO);
            client.subscribe(MQ.getTopic(), MQ.getQos());
            client.setCallback(new MqttCallback() {
                @Override
                public void connectionLost(Throwable throwable) {
                    System.out.println(throwable.getMessage());
                }

                @Override
                public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
                    System.out.println("qos"+mqttMessage.getQos());
                    System.out.println("topic:"+topic);
                    System.out.println("Payload"+new String(mqttMessage.getPayload()));


                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                    System.out.println("deliveryComplete"+ iMqttDeliveryToken.isComplete());
                }
            });
            client.getServerURI();
            return client.isConnected();
        } catch (MqttException e) {
            System.out.println("连接失败"+e.getMessage());

            return false;
        }
    }
//    可以手动获取Payload
    private byte[] getPayload() throws InterruptedException {
        final byte[][] Payload = {new String("暂无负载").getBytes()};
        CountDownLatch latch = new CountDownLatch(1);
        client.setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable throwable) {
                System.out.println(throwable.getMessage());
            }

            @Override
            public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
                System.out.println("Payload"+new String(mqttMessage.getPayload()));
                Payload[0] = mqttMessage.getPayload();
                latch.countDown();
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                System.out.println("deliveryComplete"+ iMqttDeliveryToken.isComplete());
            }
        });
        latch.await(5, TimeUnit.SECONDS);
        return Payload[0];
    }
//    发生消息

    /**
     * 这是一个MQTT消息发送类
     * @param message 发生消息
     * @param retained 是否是保留消息
     * @return
     */
    public Boolean sendMessage(String message,boolean retained){
        if(MQ.getTopic().length() == 1 || MQ.getTopic().length() > MQ.getMAXTOPIC()   )
        {
            System.out.println("====topic长度不符合规则====");
            return false;
        }
        if(MQ.getTopic().contains(" ") || MQ.getTopic().isEmpty()){
            System.out.println("====topic不能有空格或者为空====");
            return false;
        }

        try {
            client.publish(MQ.getTopic(),message.getBytes(), MQ.getQos(), retained);
            return true;
        } catch (MqttException e) {
            System.out.println("========="+e+"=========");
            return false;
        }

    }


}
